package com.neusoft.databus.policy.expression.impl;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.jexl3.JexlContext;
import org.apache.commons.jexl3.MapContext;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.google.common.collect.Lists;
import com.neusoft.bizcore.common.cache.CacheService;
import com.neusoft.bizcore.web.exception.CustomRuntimeException;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.ResourceBean;
import com.neusoft.databus.common.constant.DatabusErrorCode;
import com.neusoft.databus.common.utils.DefaultThreadFactory;
import com.neusoft.databus.policy.bean.EventBean;
import com.neusoft.databus.policy.config.DatabusPolicyProperties;
import com.neusoft.databus.policy.constants.DatabusPolicyConstant;
import com.neusoft.databus.policy.enums.PolicyType;
import com.neusoft.databus.policy.model.Expression;
import com.neusoft.databus.policy.model.Policy;
import com.neusoft.databus.policy.model.PolicyRefResource;
import com.neusoft.databus.policy.repository.PolicyRefResourceRepository;
import com.neusoft.databus.policy.repository.PolicyRepository;
import com.neusoft.databus.policy.utils.ExpressionCalculator;
import com.neusoft.databus.policy.utils.PolicyUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * 基于设备数据处理类
 *
 * @author hanbq
 * @date 2015-10-9 下午2:41:05
 */
@Slf4j
@Component("policyExpressionHandler")
public class PolicyExpressionHandlerImpl extends PolicyExpressionHandler {

    @Autowired
    private DatabusPolicyProperties policyProperties;
    @Autowired
    private PolicyRepository policyRepository;
    @Autowired
    private PolicyRefResourceRepository policyRefResourceRepository;
    @Autowired
    private CacheService cacheService;

    private static ThreadPoolExecutor dataExecutor;

    /**
     * execute处理线程
     */
    @Override
    public void execute(final CommandBean commandBean) {
        if (PolicyExpressionHandlerImpl.dataExecutor == null) {
            synchronized (PolicyExpressionHandlerImpl.class) {
                final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
                PolicyExpressionHandlerImpl.dataExecutor = new ThreadPoolExecutor(
                        Integer.valueOf(this.policyProperties.getExpressionHandlerThreadSize()),
                        Integer.valueOf(this.policyProperties.getExpressionHandlerThreadMaxSize()),
                        this.policyProperties.getExpressionHandlerKeepAliveTime(), TimeUnit.MICROSECONDS, queue,
                        new DefaultThreadFactory("policy-ex"),
                        new ThreadPoolExecutor.CallerRunsPolicy());
            }
        }
        DataHandlerRunnable runnable = new DataHandlerRunnable(commandBean);
        PolicyExpressionHandlerImpl.dataExecutor.execute(runnable);
        runnable = null;
    }

    private class DataHandlerRunnable implements Runnable {

        private CommandBean commandBean;

        public DataHandlerRunnable(final CommandBean commandBean) {
            this.commandBean = commandBean;
        }

        @Override
        public void run() {
            final String traceId = this.commandBean.generateTraceId();
            MDC.put("trace", traceId);
            // 获取该资源相关策略
            try {
                // 错误码判断
                final String errorCode = this.commandBean.getErrorCode();
                for (final ResourceBean resourceBean : this.commandBean.getResourceBean()) {
                    final Date time = this.commandBean.getCollectEndTime();
                    if (StringUtils.isNoneBlank(errorCode)) {
                        PolicyExpressionHandlerImpl.this.sendSysEvent(resourceBean, time, errorCode);
                        return;
                    } else {
                        PolicyExpressionHandlerImpl.this.sendFixSysEvent(resourceBean, time);
                    }
                    final String resourceCode = resourceBean.getCode();
                    final List<PolicyRefResource> policyRefResources =
                            PolicyExpressionHandlerImpl.this.policyRefResourceRepository
                                    .findByResourceCode(resourceCode);
                    final List<Policy> policyList = Lists.newArrayList();
                    policyRefResources.forEach(policyRefResource -> {
                        final Policy policy = policyRefResource.getPolicy();
                        if (!policy.isDisabled()) {
                            if (PolicyType.POLICY_ALARM.equals(policy.getPolicyType())) {
                                policyList.add(policy);
                            }
                        }
                    });
                    // 过滤掉不存在告警策略的设备
                    if ((policyList == null) || policyList.isEmpty()) {
                        PolicyExpressionHandlerImpl.log.debug("资源 {}无相关告警规则", resourceCode);
                    } else {
                        PolicyExpressionHandlerImpl.log.debug("资源 {}相关告警规则个数为{}", resourceCode, policyList.size());
                        policyList.forEach(policy -> {
                            PolicyExpressionHandlerImpl.this.pidHandler(resourceBean, policy, time);
                        });
                    }
                }

            } catch (final Exception e) {
                PolicyExpressionHandlerImpl.log.error("策略告警解析异常：{}", e);
            } finally {
                this.commandBean = null;
                MDC.clear();
            }
        }
    }

    /**
     * 系统告警
     *
     * @param resourceBean
     * @param time
     * @throws Exception
     */
    private void sendSysEvent(final ResourceBean resourceBean, final Date time, final String errorCode)
            throws Exception {
        final Policy policy = this.policyRepository.findByCode(errorCode);
        if (null == policy) {
            PolicyExpressionHandlerImpl.log.warn("can not find policy definition of code: {}", errorCode);
            return;
        }
        if (policy.isDisabled()) {
            PolicyExpressionHandlerImpl.log.warn("系统告警被禁用：{}", errorCode);
            return;
        }
        // 频率判断
        if (StringUtils.isNotBlank(policy.getCriterion())) {
            final Map<String, Object> criterion = JsonUtils.jsonToMap(policy.getCriterion());
            if (criterion.containsKey(DatabusPolicyConstant.FREQUENCY)) {
                final String frequencyKey = PolicyUtil.getFrequencyKey(resourceBean.getCode(), policy);
                if (!this.judgmentFrequency(frequencyKey,
                        JsonUtils.jsonToMap(criterion.get(DatabusPolicyConstant.FREQUENCY).toString()),
                        time.getTime())) {
                    return;
                }
            }
        }
        final EventBean eventBean = new EventBean();
        eventBean.setExpressionResult(true);
        eventBean.setResourceBean(resourceBean);
        eventBean.setExpressionTrueSquence(null);
        final List<Policy> policies = Lists.newArrayList();
        policies.add(policy);
        eventBean.setPolicies(policies);
        eventBean.setTime(time);
        this.sendResult(eventBean);
        PolicyExpressionHandlerImpl.log.info("系统告警:resourceCode:{},policyCode:{}", resourceBean.getCode(),
                policy.getCode());
    }

    /**
     * 系统告警解除
     *
     * @param resourceBean
     * @param time
     */
    private void sendFixSysEvent(final ResourceBean resourceBean, final Date time) {
        final EventBean eventBean = new EventBean();
        eventBean.setExpressionResult(false);
        eventBean.setResourceBean(resourceBean);
        eventBean.setExpressionTrueSquence(null);
        final List<PolicyType> types = Lists.newArrayList();
        types.add(PolicyType.DEVICE_FAULT);
        types.add(PolicyType.GATEWAY_DISCONNECT);
        types.add(PolicyType.SYSTEM);
        List<Policy> policies = Lists.newArrayList();
        policies = this.policyRepository.findByPolicyTypeIn(types);
        eventBean.setPolicies(policies);
        eventBean.setTime(time);
        this.sendResult(eventBean);
    }

    /**
     * 指标公式解析类
     *
     * @param commandBean
     * @param policy
     * @param time
     */
    private void pidHandler(ResourceBean resourceBean, final Policy policy, final Date time) {
        final List<Expression> expList = policy.getExpressions();
        final Long policyId = policy.getId();
        // 升序排列表达式
        Collections.sort(expList,
                (exp1, exp2) -> new Double(exp1.getSequence()).compareTo(new Double(exp2.getSequence())));
        EventBean eventBean = null;
        List<Integer> expressionTrueSquence = null;
        StringBuffer expResultSB = null;
        final JexlContext jexlContext = new MapContext();
        try {
            eventBean = new EventBean();
            expressionTrueSquence = new ArrayList<>();
            expResultSB = new StringBuffer();
            for (Expression expEntity : expList) {
                if (expEntity.getLinkSymbol() != null) {
                    expResultSB.append(expEntity.getLinkSymbol());
                }
                Map<String, Object> exp = JsonUtils.jsonToMap(expEntity.getExpressionInfo());
                Boolean result = false;
                // 根据expType解析jexl表达式
                if (expEntity.getType() != null) {
                    switch (expEntity.getType()) {
                    case COMPARE:
                        // 指标比较
                        result = this.policyExpressionParser.valueCompare(exp, resourceBean);
                        break;
                    case CHANGE:
                        // 指标变化
                        result = this.policyExpressionParser.valueChange(exp, resourceBean, policyId);
                        break;
                    case PERCENTAGECHANGE:
                        // 指标波动率
                        result = this.policyExpressionParser.valuePercentageChange(exp, resourceBean, policyId);
                        break;
                    case SCOPECHANGE:
                        // 指标变化大小
                        result = this.policyExpressionParser.valueScopeChange(exp, resourceBean, policyId);
                        break;
                    case CUSTOMIZE:
                        // 自定义
                        result = this.policyExpressionParser.custom(exp, resourceBean, policyId);
                        break;
                    default:
                        PolicyExpressionHandlerImpl.log.error("不存在该数据{}解析方法", expEntity.getType());
                        break;
                    }
                }
                expressionTrueSquence.add(expEntity.getSequence());
                expResultSB.append(result.toString());
                jexlContext.set(String.valueOf(expEntity.getSequence()), result);
                exp = null;
                expEntity = null;
            }
            final boolean resultTotal = ExpressionCalculator.getJexlCalculatorResult(expResultSB.toString(),
                    jexlContext);
            // 频率判断
            if (resultTotal && StringUtils.isNotBlank(policy.getCriterion())) {
                final Map<String, Object> criterion = JsonUtils.jsonToMap(policy.getCriterion());
                if (criterion.containsKey(DatabusPolicyConstant.FREQUENCY)) {
                    final String frequencyKey = PolicyUtil.getFrequencyKey(resourceBean.getCode(), policy);
                    if (!this.judgmentFrequency(frequencyKey,
                            JsonUtils.jsonToMap(criterion.get(DatabusPolicyConstant.FREQUENCY).toString()),
                            time.getTime())) {
                        return;
                    }
                }
            }
            eventBean.setExpressionResult(resultTotal);
            eventBean.setResourceBean(resourceBean);
            eventBean.setExpressionTrueSquence(expressionTrueSquence);
            final List<Policy> policies = Lists.newArrayList();
            policies.add(policy);
            eventBean.setPolicies(policies);
            eventBean.setTime(time);
            this.sendResult(eventBean);
        } catch (final Exception e) {
            PolicyExpressionHandlerImpl.log.error("策略解析处理异常：{}", e);
            return;
        } finally {
            expressionTrueSquence = null;
            expResultSB = null;
            eventBean = null;
            resourceBean = null;
        }
    }

    /**
     * 触发频率判断方法
     *
     * @param frequencyKey
     * @param frequency
     * @param time
     * @return
     * @throws Exception
     */
    @SuppressWarnings("unchecked")
    protected boolean judgmentFrequency(final String frequencyKey, final Map<String, Object> frequency, final Long time)
            throws Exception {
        final String cacheName = DatabusPolicyConstant.POLICY_FREQUENCY_CACHE_NAME;
        if ((frequency.get(DatabusPolicyConstant.TIME) == null)
                || (frequency.get(DatabusPolicyConstant.TIMES) == null)) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_406220);
        }
        final Long timeFrequency = Long.parseLong(frequency.get(DatabusPolicyConstant.TIME).toString());
        final Integer timesFrequency = Integer.valueOf(frequency.get(DatabusPolicyConstant.TIMES).toString());

        if (timesFrequency < 2) {
            return true;
        }
        boolean result = false;
        List<Long> timeList = Lists.newLinkedList();
        if (this.cacheService.isInCache(cacheName, frequencyKey)) {
            timeList = (List<Long>) this.cacheService.get(cacheName, frequencyKey);
        }
        timeList.add(time);
        if ((time - timeList.get(0)) > timeFrequency) {
            timeList.remove(0);
            this.cacheService.put(cacheName, frequencyKey, timeList);
            result = false;
        }
        if (timeList.size() < timesFrequency) {
            this.cacheService.put(cacheName, frequencyKey, timeList);
            result = false;
        } else {
            this.cacheService.remove(cacheName, frequencyKey);
            result = true;
        }
        timeList = null;
        return result;
    }
}
